package com.xinqi.common.rocketmq.processor;

import com.xinqi.common.rocketmq.annotation.MqConsumer;
import com.xinqi.common.rocketmq.dto.RocketMqBean;
import com.xinqi.common.rocketmq.util.MqHelpUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.MethodIntrospector;
import org.springframework.core.env.Environment;
import org.springframework.lang.Nullable;

import java.lang.reflect.Method;
import java.util.Map;

/**
 * @author linkai
 * @date 2023/11/18
 */
@Slf4j
public class MqConsumerPostProcessor implements BeanPostProcessor, EnvironmentAware {

    private Environment environment;
    @Autowired
    private RocketMqBean rocketMqBean;

    @Override
    public Object postProcessAfterInitialization(@Nullable Object bean, @Nullable String beanName)
        throws BeansException {
        if (bean == null) {
            return null;
        }
        // 如果bean上加了注解，且bean继承了BaseMqConsumer类
        // 获取一个代理对象的最终对象类型
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
        if (bean instanceof MessageListener && clazz.isAnnotationPresent(MqConsumer.class)) {
            try {
                buildMqConsumer(beanName, (MessageListener) bean, clazz.getAnnotation(MqConsumer.class));
            } catch (ClientException e) {
                throw new RuntimeException(e);
            }
            return bean;
        }

        // 处理方法上的MqConsumer注解
        Map<Method, MqConsumer> annotatedMethods = MethodIntrospector.selectMethods(clazz,
            (MethodIntrospector.MetadataLookup<MqConsumer>) method -> method.getAnnotation(MqConsumer.class));
        if (!annotatedMethods.isEmpty()) {
            annotatedMethods.forEach((method, mqConsumer) -> processConsumer(mqConsumer, method, bean, beanName));
        }
        return bean;
    }

    private void processConsumer(MqConsumer mqConsumer, Method method, Object bean, String beanName) {
        try {
            MessageListener consumer = createMqConsumer(bean, method);
            String defaultGroupName = beanName + "_" + method.getName();
            buildMqConsumer(defaultGroupName, consumer, mqConsumer);
        } catch (ClientException e) {
            throw new RuntimeException(e);
        }
    }

    private <T> MessageListener createMqConsumer(Object target, Method method) {
        Method invocableMethod = AopUtils.selectInvocableMethod(method, target.getClass());
        return new MqConsumerListener<>(target, invocableMethod, (Class<T>) method.getParameterTypes()[0]);
    }

    private void buildMqConsumer(String defaultGroupName, MessageListener listener, MqConsumer mqConsumer)
        throws ClientException {
        String groupName = mqConsumer.groupName();
        if (StringUtils.isBlank(mqConsumer.groupName())) {
            groupName = defaultGroupName;
        }
        if (StringUtils.isBlank(mqConsumer.topic())) {
            throw new IllegalArgumentException("MQ Consumer create failed!, topic is empty, class = " + listener.getClass());
        }
        String topic = getTopic(mqConsumer.topic().trim());
        MqHelpUtils.createPushConsumer(topic, groupName, mqConsumer.tag(), mqConsumer.consumptionThreadCount(),
            rocketMqBean, listener);
        log.info("mq consumer start,  NameSrvAddr:{} GroupName:{}, Topic:{}",
            rocketMqBean.getEndpoints(),
            mqConsumer.groupName(), topic);
    }

    private String getTopic(String el) {
        if (el.startsWith("${")) {
            String paramKey = el.substring(2, el.length() - 1);
            return this.environment.getProperty(paramKey);
        }
        return el;
    }

    /**
     * Set the {@code Environment} that this component runs in.
     *
     * @param environment
     */
    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }
}
